package com.tc.property.config.zk;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.google.common.collect.Maps;

import com.tc.property.config.AbstractConfig;
import com.tc.property.config.handler.IConfigSourceChangedHandler;
import com.tc.property.config.model.ConfigItem;
import com.tc.property.util.Constants;
import com.tc.property.util.Utils;
import com.tc.zookeeper.curator.EnhancedCuratorFramework;
import com.tc.zookeeper.curator.EnhancedCuratorFrameworkFactory;
import com.tc.zookeeper.curator.EnhancedNodeCacheListener;
import com.tc.zookeeper.exception.EnhancedCuratorException;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.utils.ZKPaths;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.management.MBeanServer;
import javax.management.ObjectName;
import java.io.*;
import java.lang.management.ManagementFactory;
import java.nio.charset.Charset;
import java.util.*;
import java.util.Map.Entry;

public class CfgCenterConfig extends AbstractConfig implements CfgCenterConfigMBean {

	private static Logger log = LoggerFactory.getLogger(CfgCenterConfig.class);

	private static final String APPS_SVR_CFGBACKUP = "/apps/svr/cfgbackup";

	private EnhancedCuratorFramework zkClient;

	private final String location;

	private final String domain;

	private String targetConnectString = "";

	private Long globalSessionID = 0L;

	private List<IConfigSourceChangedHandler> handlers;

	Boolean isReportEvent = true;

	private boolean isFirstGetConfig = true;

	/**
	 * source like "cfgcenter:vip.lusong.com:/artifact/com.vip.venus/cebstestzk-webapp/1.0.0/config"
	 * @param source
	 * ${app.name}:/config/${app.name}
	 */
	public CfgCenterConfig(String source, String zkConnectString, Boolean ignoreResourceNotFound) {
		super(source, ignoreResourceNotFound);
		source = source.replaceFirst(Constants.CFGCENRTER_PREFIX, Constants.EMPTY_STRING);
		String[] paths = StringUtils.split(source, ':');
		if (paths.length == 2) {
			domain = paths[0];
			location = paths[1];
		} else {
			domain = null;
			location = null;
			if (!ignoreResourceNotFound) {
				throw new CfgCenterConfigException(String.format("wrong source:%s", this.source));
			}
		}
		try {
			try {
				targetConnectString = EnhancedCuratorFrameworkFactory.getTargetConnectStringByConnectDomain(zkConnectString, domain);
			} catch (EnhancedCuratorException e1) {
			}
			if (StringUtils.isBlank(targetConnectString)) {
				log.info(">>>>>>use zkConnectString:" + zkConnectString);
				targetConnectString = zkConnectString;
			}
			if (StringUtils.isBlank(targetConnectString)) {
				log.error("Can not find VIP_CFGCENTER_ZK_CONNECTION config.");
				throw new CfgCenterConfigException("Please check your VIP_CFGCENTER_ZK_CONNECTION config.");
			}
			try {
				log.info(">>>>>>final CfgCenterConfig targetConnectString:" + targetConnectString);
				zkClient = EnhancedCuratorFrameworkFactory.getStartedClientByConnectString(targetConnectString);
				buildZkListener();
				registerMBean();
			} catch (EnhancedCuratorException e) {
				throw e;
			}
		} catch (Exception e) {
			if (!ignoreResourceNotFound) {
				throw new CfgCenterConfigException(e);
			}
		}
	}

	public CfgCenterConfig(String source, Boolean ignoreResourceNotFound, Boolean isReportEvent) {
		super(source, ignoreResourceNotFound);
		this.isReportEvent = isReportEvent;
		source = source.replace(Constants.CFGCENRTER_PREFIX, Constants.EMPTY_STRING);
		String[] paths = StringUtils.split(source, ':');
		if (paths.length == 2) {
			domain = paths[0];
			location = paths[1];
		} else {
			domain = null;
			location = null;
			if (!ignoreResourceNotFound) {
				throw new CfgCenterConfigException(String.format("wrong source:%s", this.source));
			}
		}
		try {
			try {
				targetConnectString = EnhancedCuratorFrameworkFactory.getTargetConnectString(location);
			} catch (EnhancedCuratorException e1) {
			}
			if (StringUtils.isBlank(targetConnectString)) {
				log.error("Can not find VIP_CFGCENTER_ZK_CONNECTION config.");
				throw new CfgCenterConfigException("Please check your VIP_CFGCENTER_ZK_CONNECTION config.");
			}
			try {
				log.info(">>>>>>final CfgCenterConfig targetConnectString:" + targetConnectString);
				zkClient = EnhancedCuratorFrameworkFactory.getStartedClientByConnectString(targetConnectString);
				buildZkListener();
				registerMBean();
			} catch (EnhancedCuratorException e) {
				throw e;
			}
		} catch (Exception e) {
			if (!ignoreResourceNotFound) {
				throw new CfgCenterConfigException(e);
			}
		}
	}

	private void buildZkListener() {
		zkClient.getConnectionStateListenable().addListener(new ConnectionStateListener() {
			@Override
			public void stateChanged(CuratorFramework client, ConnectionState connectionState) {
				stateChangedProcess(connectionState);
			}
		});
	}

	public void stateChangedProcess(ConnectionState connectionState) {
		//去除不必要的状态判断和
		switch (connectionState) {
			case CONNECTED:
				if (!isFirstGetConfig){//一开始连接不上，后来连接上后更新数据
					getConfig(handlers);
				}
				break;
			case RECONNECTED:
				// 重连之后，无论watch是否注册，强制重新注册
				getConfig(handlers);
				break;
			default:
				break;
		}
	}

	public Map<String, ConfigItem> loadConfigsFromBackupFile() {
		Map<String, ConfigItem> contextItemMap = Maps.newHashMap();
		String filePath = getBackupFileName();
		if (StringUtils.isBlank(filePath)){
			return contextItemMap;
		}
		File file = new File(filePath);
		if (file.exists()) {
			log.info(String.format("load configs from backup file : %s",filePath));
			contextItemMap =  readConfigsFromFile(filePath);
		}
		return contextItemMap;
	}

	public Map<String, ConfigItem> readConfigsFromFile(String filePath) {
		Map<String, ConfigItem> contextItemMap = Maps.newHashMap();
		try {
			Properties props = new Properties();
			FileInputStream fileInputStream = new FileInputStream(filePath);
			InputStreamReader inputStreamReader = new InputStreamReader(fileInputStream);
			BufferedReader bufferedReader = new BufferedReader(inputStreamReader);
			props.load(bufferedReader);
			bufferedReader.close();
			inputStreamReader.close();
			fileInputStream.close();
			Set<Entry<Object, Object>> entrySet = props.entrySet();
			for (Entry<Object, Object> entry : entrySet) {
				String key = entry.getKey() + "";
				contextItemMap.put(key,new ConfigItem(key, entry.getValue() + "", source, 0L, System.currentTimeMillis()));
			}
		} catch (Exception e) {
			log.error("updateHandlerFromFile ," + e.getMessage());
		}
		return contextItemMap;
	}

	@Override
	public Map<String, ConfigItem> getConfig(List<IConfigSourceChangedHandler> handlers) {
		try {
			if (null == this.handlers) {
				this.handlers = handlers;
			}
			Map<String, ConfigItem> result = executeHandler();
			if(isReportEvent){
				reportEvent(globalSessionID, targetConnectString);
			}
			return result;
		} finally {
			//注册watch Listener
			doWatchNode();
			isFirstGetConfig = false;
		}
	}

	private void doWatchNode() {
		if (null == zkClient || !zkClient.getZookeeperClient().isConnected()){
			return;
		}
		if (checkNodeExist(this.location)) {
			watchNode();
		} else{
			watchParentNode(); //在父节点修改是实现对子节点的监控，当子节点修改时调用watchNode()
		}
	}

	private boolean checkNodeExist(String path) {
		Stat stat = null;
		try {
			stat = zkClient.checkExists().forPath(path);
		} catch (Exception e) {
			stat = null;
		}
		return stat != null;
	}

	public void watchParentNode() {
		try {
			final String zkParentPath = CfgCenterUtils.getParentZkPath(this.location);
			final PathChildrenCache cache = new PathChildrenCache(zkClient, zkParentPath, true);
			cache.start();
			cache.getListenable().addListener(new PathChildrenCacheListener() {
				@Override
				public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
					PathChildrenCacheEvent.Type eventType = event.getType();
					if (eventType == PathChildrenCacheEvent.Type.CHILD_ADDED) {
						String path = event.getData().getPath();
						if (path.equals(location)) {
							try {
								watchNode();
							} finally {
								executeHandler();
								cache.close();
							}
						}
					}
				}
			});
		} catch (Exception e) {
			log.error("doPathCache" + e.getMessage());
		}
	}

	private void watchNode() {
		try {
			EnhancedNodeCacheListener listener = new EnhancedNodeCacheListener() {
				public void nodeChanged() throws Exception {
					nodeChangeProcess();
				}
			};
			zkClient.watchNode(location, listener);
		} catch (Exception e) {
			log.error("addWatchListener" + e.getMessage());
		}
	}

	public void nodeChangeProcess() {
		executeHandler();
		if(isReportEvent){
			reportEvent(globalSessionID, targetConnectString);
		}
	}

	private Map<String, ConfigItem> executeHandler() {
		Map<String, ConfigItem> configMap = Maps.newHashMap();
		try {
			Stat stat = new Stat();
			byte[] zkData = zkClient.getData().storingStatIn(stat).forPath(location);
			final long dataVersion = stat.getVersion();
			String data = new String(zkData, Charset.forName("UTF-8"));
			if (StringUtils.isNotBlank(data)) {
				/*ZkUpdateProtocol protocol = ZkUpdateProtocol.fromJson(data, ZkUpdateProtocol.class);
				globalSessionID = protocol.getHeader().getGlobalSessionID();
				Map<String, String> cacheMap = CfgCenterUtils.decode(data);*/
				Map<String, String> cacheMap = JSONObject.parseObject(data, new TypeReference<Map<String, String>>(){});
				for (Entry<String, String> entry : cacheMap.entrySet()) {
					String key = entry.getKey();
					String value = entry.getValue();
					ConfigItem configData = new ConfigItem(key, value, source, dataVersion, globalSessionID);
					configMap.put(key, configData);
				}
				//获取到zk数据后，配置写入备份文件，不包括移除的key
				writeBackUpFile(configMap);
			}
			else{
				log.warn(String.format("node:%s，data为空！", location));
				//节点为空，从备份文件中获取
				configMap = loadConfigsFromBackupFile();
			}
		} catch (Throwable e) {
			log.warn(String.format("%s on %s", e.getMessage(), targetConnectString) + "," + e.getMessage());
			//zk获取不到数据则从备份文件中获取
			configMap = loadConfigsFromBackupFile();
		} finally {
			if (handlers != null) {
				for (final IConfigSourceChangedHandler handler : handlers) {
					handler.update(configMap);
				}
			}
		}
		return configMap;
	}

	public void writeBackUpFile(Map<String, ConfigItem> configMap) {
		String backupFilePath = "";
		try {
			backupFilePath = getBackupFileName();
			if (StringUtils.isBlank(backupFilePath)){
				return;
			}
			forceMkDirectory(backupFilePath);
			/*String propertiesString = propertiesToString(configMap);
			if (StringUtils.isNotBlank(propertiesString)) {
				FileUtils.write(new File(backupFilePath), propertiesString, "utf-8");
			}*/
			Properties pro = mapToProperties(configMap);
			if(!pro.isEmpty()){
				FileOutputStream fileOutputStream = new FileOutputStream(new File(backupFilePath));
				OutputStreamWriter outputStreamWriter = new OutputStreamWriter(fileOutputStream);
				BufferedWriter bufferedWriter = new BufferedWriter(outputStreamWriter);
				pro.store(bufferedWriter, "配置数据");
				bufferedWriter.close();
				outputStreamWriter.close();
				fileOutputStream.close();
			}
		} catch (Exception e) {
			log.warn("write backup file failed." + backupFilePath + "," + e.getMessage());
			if(isReportEvent){
				reportEvent(globalSessionID, targetConnectString);
			}

		}
	}

	public String getBackupFileName(){
		if (zkClient == null){
			return "";
		}
		String zkPath = ZKPaths.makePath(zkClient.getNamespace(), location);
		zkPath = APPS_SVR_CFGBACKUP + zkPath;
		return zkPath+".properties";
	}

	public void forceMkDirectory(String zkPath) throws IOException {
		String parentDir = zkPath.substring(0, StringUtils.lastIndexOf(zkPath, "/"));
		File parentDirFile = new File(parentDir);
		if (!parentDirFile.exists()) {
			FileUtils.forceMkdir(parentDirFile);
		}
	}

	public static String propertiesToString(Map<String, ConfigItem> configMap) {
		String result = "";
		if (configMap != null) {
			for (Entry<String, ConfigItem> entry : configMap.entrySet()) {
				String key = entry.getKey();
				ConfigItem item = entry.getValue();
				if (item != null) {
					result += (key + "=" + item.getValue() + "\n");
				}
			}
		}
		return result;
	}

	private Properties mapToProperties(Map<String, ConfigItem> configMap){
		Properties pro = new Properties();
		if (configMap != null) {
			for (Entry<String, ConfigItem> entry : configMap.entrySet()) {
				String key = entry.getKey();
				ConfigItem item = entry.getValue();
				if (item != null) {
					pro.put(key, item.getValue());
				}
			}
		}
		return pro;
	}

	private void reportEvent(Long globalSessionID, String targetConnectString) {
		log.info("start CfgCenterConfig reportEvent....");
		try {
			/*JSONObject json = new JSONObject();
			json.put("domain", domain);
			String localAddress = com.vip.hermes.core.util.HermesUtil.localIp4Str();
			json.put("ipAddress", localAddress);
			String localHostName = Utils.getLocalHostName();
			json.put("hostName", localHostName);
			json.put("globalSessionID", globalSessionID);
			json.put("targetConnectString", targetConnectString);

			Map<String, Object> customFields = new HashMap<String, Object>();
			customFields.put("domain",domain);
			customFields.put("localAddress",localAddress);
			customFields.put("localHostName",localHostName);
			customFields.put("globalSessionID",globalSessionID + "");
			customFields.put("targetConnectString",targetConnectString);

			com.vip.hermes.common.model.event.EventDetailData detail = new com.vip.hermes.common.model.event.EventDetailData();
			detail.setMessage(json.toJSONString());
			detail.setName("CONFIG.CFG.EVENT");
			detail.setTitle("CONFIG.CFG.ZKDATA");
			detail.setType("CONFIG.CFG.DOMAIN");
			detail.setCustomFields(customFields);
			com.vip.hermes.event.reporter.EventReportHelper.report(detail);*/
		} catch (Throwable e) {
			log.warn("CfgCenterConfig reportEvent error :" + e.getMessage());
		}
	}

	public void registerMBean() {
		synchronized (this) {
			MBeanServer mbeanServer = ManagementFactory.getPlatformMBeanServer();
			try {
				ObjectName objectName = new ObjectName("com.vip.venus.config.impl:type=" + this.toString() + "|"
						+ targetConnectString.replaceAll(":", "_").replaceAll(",", "-"));
				if (!mbeanServer.isRegistered(objectName)) {
					mbeanServer.registerMBean(this, objectName);
					log.info("com.vip.venus.config.impl:type=" + this.toString() + "|"
							+ targetConnectString.replaceAll(":", "_").replaceAll(",", "-")
							+ " registered successfully");
				}
			} catch (Exception e) {
				log.error("registerMBean error:" + e.getMessage());
			}
		}
	}

	@Override
	public void reloadCfgData() {
		executeHandler();
	}

}
